package com.sure0000.hudi;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @desc
 * @Author xuyouchang
 * @Date 2021/10/21
 */
public class Kafka2Hudi {
    public static void main(String[] args) {
        Configuration configuration = new Configuration();
        configuration.setString(RestOptions.BIND_PORT, "8081-8083");
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(configuration);
        env.enableCheckpointing(5000L);
        env.setStateBackend(new FsStateBackend("file:///Users/xuyouchang/flink/state"));
        EnvironmentSettings bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
        StreamTableEnvironment bsTableEnv = StreamTableEnvironment.create(env, bsSettings);


        bsTableEnv.executeSql("create table kafka_table (id int, gender string, name string) " +
                "with ('connector'='kafka', 'topic' = 'xyc_test_json_02', 'properties.bootstrap.servers' = '172.17.1.101:9092'," +
                        "'properties.group.id' = 'test03','scan.startup.mode' = 'earliest-offset','format' = 'json')"
                );

        bsTableEnv.executeSql(
                "create table hudi_table (id int, gender string, name string, ts TIMESTAMP(3),PRIMARY KEY (`id`) NOT ENFORCED) partitioned by (gender)" +
                "with ('connector'='hudi', 'path'='hdfs://172.17.1.105:8020/hudi/xyc/test_json_04', " +
                        "'table.type'='MERGE_ON_READ')");


        Table table2 = bsTableEnv.sqlQuery("select id, gender, name, current_timestamp as ts from kafka_table");
        table2.executeInsert("hudi_table");

    }
}
